package org.nehc.algorithm.redissonutils;

import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.*;
import org.redisson.api.options.KeysScanOptions;
import org.redisson.api.queue.*;

import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * Redisson 队列和消息操作工具类
 * 提供队列操作和发布订阅功能
 *
 * @author system
 */
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
@SuppressWarnings(value = {"unchecked", "rawtypes"})
public class RedissonQueueUtils {

    private static final RedissonClient CLIENT = RedissonUtils.getClient();

    // ==================== 队列操作 ====================

    /**
     * 通用销毁队列 所有阻塞监听 报错(不支持延迟队列)
     *
     * @param queueName 队列名称
     * @return true=销毁成功；false=销毁失败
     */
    public static <T> boolean destroyQueue(String queueName) {
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        return queue.delete();
    }

    /**
     * 添加普通队列数据
     *
     * @param queueName 队列名
     * @param data      数据
     * @return true=添加成功；false=添加失败
     */
    public static <T> boolean addQueueObject(String queueName, T data) {
        destroyQueue(queueName);
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        return queue.offer(data);
    }

    /**
     * 添加队列数据（不销毁现有队列）
     *
     * @param queueName 队列名
     * @param data      数据
     * @return true=添加成功；false=添加失败
     */
    public static <T> boolean offerQueueObject(String queueName, T data) {
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        return queue.offer(data);
    }

    /**
     * 从队列头部获取并移除元素（非阻塞）
     *
     * @param queueName 队列名
     * @return 队列头部元素，如果队列为空则返回null
     */
    public static <T> T pollQueueObject(String queueName) {
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        return queue.poll();
    }

    /**
     * 从队列头部获取但不移除元素
     *
     * @param queueName 队列名
     * @return 队列头部元素，如果队列为空则返回null
     */
    public static <T> T peekQueueObject(String queueName) {
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        return queue.peek();
    }

    /**
     * 从队列头部获取并移除元素（阻塞）
     *
     * @param queueName 队列名
     * @return 队列头部元素
     * @throws InterruptedException 中断异常
     */
    public static <T> T takeQueueObject(String queueName) throws InterruptedException {
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        return queue.take();
    }

    /**
     * 获取队列大小
     *
     * @param queueName 队列名
     * @return 队列大小
     */
    public static int getQueueSize(String queueName) {
        RBlockingQueue<?> queue = CLIENT.getBlockingQueue(queueName);
        return queue.size();
    }

    /**
     * 检查队列是否为空
     *
     * @param queueName 队列名
     * @return true=队列为空；false=队列不为空
     */
    public static boolean isQueueEmpty(String queueName) {
        RBlockingQueue<?> queue = CLIENT.getBlockingQueue(queueName);
        return queue.isEmpty();
    }

    /**
     * 清空队列
     *
     * @param queueName 队列名
     */
    public static void clearQueue(String queueName) {
        RBlockingQueue<?> queue = CLIENT.getBlockingQueue(queueName);
        queue.clear();
    }

    /**
     * 订阅队列(可订阅所有实现类 例如: 延迟 优先 有界 等)
     *
     * @param queueName 队列名
     * @param consumer  消费者处理函数
     */
    public static <T> void subscribeBlockingQueue(String queueName, Consumer<T> consumer) {
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        queue.subscribeOnElements(consumer);
    }

    /**
     * 检查队列是否存在
     *
     * @param queueName 队列名
     * @return true=队列存在；false=队列不存在
     */
    public static boolean isQueueExists(String queueName) {
        RBlockingQueue<?> queue = CLIENT.getBlockingQueue(queueName);
        return queue.isExists();
    }

    // ==================== 发布订阅操作 ====================

    /**
     * 发布通道消息
     *
     * @param channelKey 通道key
     * @param msg        发送数据
     * @param consumer   自定义处理
     */
    public static <T> void publish(String channelKey, T msg, Consumer<T> consumer) {
        RTopic topic = CLIENT.getTopic(channelKey);
        topic.publish(msg);
        consumer.accept(msg);
    }

    /**
     * 发布通道消息
     *
     * @param channelKey 通道key
     * @param msg        发送数据
     * @return 接收到消息的订阅者数量
     */
    public static <T> long publish(String channelKey, T msg) {
        RTopic topic = CLIENT.getTopic(channelKey);
        return topic.publish(msg);
    }

    /**
     * 订阅通道接收消息
     *
     * @param channelKey 通道key
     * @param clazz      消息类型
     * @param consumer   自定义处理
     * @return 监听器ID
     */
    public static <T> int subscribe(String channelKey, Class<T> clazz, Consumer<T> consumer) {
        RTopic topic = CLIENT.getTopic(channelKey);
        return topic.addListener(clazz, (channel, msg) -> consumer.accept(msg));
    }

    /**
     * 订阅通道接收消息（带通道信息）
     *
     * @param channelKey 通道key
     * @param clazz      消息类型
     * @param consumer   自定义处理（包含通道和消息）
     * @return 监听器ID
     */
    public static <T> int subscribeWithChannel(String channelKey, Class<T> clazz,
                                               java.util.function.BiConsumer<String, T> consumer) {
        RTopic topic = CLIENT.getTopic(channelKey);
        return topic.addListener(clazz, (channel, message) -> consumer.accept(channel.toString(), message));
    }

    /**
     * 取消订阅
     *
     * @param channelKey 通道key
     * @param listenerId 监听器ID
     */
    public static void unsubscribe(String channelKey, int listenerId) {
        RTopic topic = CLIENT.getTopic(channelKey);
        topic.removeListener(listenerId);
    }

    /**
     * 取消所有订阅
     *
     * @param channelKey 通道key
     */
    public static void unsubscribeAll(String channelKey) {
        RTopic topic = CLIENT.getTopic(channelKey);
        topic.removeAllListeners();
    }

    /**
     * 获取通道的订阅者数量
     *
     * @param channelKey 通道key
     * @return 订阅者数量
     */
    public static long countSubscribers(String channelKey) {
        RTopic topic = CLIENT.getTopic(channelKey);
        return topic.countSubscribers();
    }

    /**
     * 获取所有活跃的通道名称
     *
     * @return 活跃的通道名称列表
     */
    public static Iterable<String> getChannelNames() {
        return CLIENT.getKeys().getKeys(KeysScanOptions.defaults().pattern("__keyspace@*"));
    }

    /**
     * 检查通道是否存在订阅者
     *
     * @param channelKey 通道key
     * @return true=存在订阅者；false=不存在订阅者
     */
    public static boolean hasSubscribers(String channelKey) {
        return countSubscribers(channelKey) > 0;
    }

    // ==================== 可靠队列操作 (RReliableQueue) ====================

    /**
     * 获取可靠队列
     *
     * @param queueName 队列名
     * @return 可靠队列对象
     */
    public static <T> RReliableQueue<T> getReliableQueue(String queueName) {
        return CLIENT.getReliableQueue(queueName);
    }

    /**
     * 设置可靠队列配置
     *
     * @param queueName 队列名
     * @param config    队列配置
     * @return true=设置成功；false=配置已存在
     */
    public static boolean setReliableQueueConfig(String queueName, QueueConfig config) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        return queue.setConfigIfAbsent(config);
    }

    /**
     * 向可靠队列添加消息
     *
     * @param queueName 队列名
     * @param data      数据
     * @return 消息对象
     */
    public static <T> Message<T> addReliableMessage(String queueName, T data) {
        RReliableQueue<T> queue = CLIENT.getReliableQueue(queueName);
        return queue.add(QueueAddArgs.messages(MessageArgs.payload(data)));
    }

    /**
     * 从可靠队列获取消息
     *
     * @param queueName 队列名
     * @return 消息对象，如果队列为空则返回null
     */
    public static <T> Message<T> pollReliableMessage(String queueName) {
        RReliableQueue<T> queue = CLIENT.getReliableQueue(queueName);
        return queue.poll();
    }

    /**
     * 确认消息处理成功
     *
     * @param queueName 队列名
     * @param messageId 消息ID
     */
    public static void acknowledgeMessage(String queueName, String messageId) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        queue.acknowledge(QueueAckArgs.ids(messageId));
    }

    /**
     * 否定确认消息（重新投递）
     *
     * @param queueName 队列名
     * @param messageId 消息ID
     */
    public static void negativeAcknowledgeMessage(String queueName, String messageId) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        queue.negativeAcknowledge(QueueNegativeAckArgs.failed(messageId));
    }

    /**
     * 拒绝消息（移动到死信队列）
     *
     * @param queueName 队列名
     * @param messageId 消息ID
     */
    public static void rejectMessage(String queueName, String messageId) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        queue.negativeAcknowledge(QueueNegativeAckArgs.rejected(messageId));
    }

    /**
     * 批量确认消息
     *
     * @param queueName  队列名
     * @param messageIds 消息ID数组
     */
    public static void acknowledgeMessages(String queueName, String... messageIds) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        queue.acknowledge(QueueAckArgs.ids(messageIds));
    }

    /**
     * 批量否定确认消息（重新投递）
     *
     * @param queueName  队列名
     * @param messageIds 消息ID数组
     */
    public static void negativeAcknowledgeMessages(String queueName, String... messageIds) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        queue.negativeAcknowledge(QueueNegativeAckArgs.failed(messageIds));
    }

    /**
     * 批量拒绝消息（移动到死信队列）
     *
     * @param queueName  队列名
     * @param messageIds 消息ID数组
     */
    public static void rejectMessages(String queueName, String... messageIds) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        queue.negativeAcknowledge(QueueNegativeAckArgs.rejected(messageIds));
    }

    /**
     * 检查可靠队列是否为空
     *
     * @param queueName 队列名
     * @return true=队列为空；false=队列不为空
     */
    public static boolean isReliableQueueEmpty(String queueName) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        return queue.isEmpty();
    }

    /**
     * 获取可靠队列大小
     *
     * @param queueName 队列名
     * @return 队列大小
     */
    public static int getReliableQueueSize(String queueName) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        return queue.size();
    }

    /**
     * 清空可靠队列
     *
     * @param queueName 队列名
     */
    public static void clearReliableQueue(String queueName) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        queue.clear();
    }

    /**
     * 获取可靠队列中未确认消息数量
     *
     * @param queueName 队列名
     * @return 未确认消息数量
     */
    public static int getUnacknowledgedMessageCount(String queueName) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        return queue.countUnacknowledgedMessages();
    }

    /**
     * 获取可靠队列中延迟消息数量
     *
     * @param queueName 队列名
     * @return 延迟消息数量
     */
    public static int getDelayedMessageCount(String queueName) {
        RReliableQueue<?> queue = CLIENT.getReliableQueue(queueName);
        return queue.countDelayedMessages();
    }

    // ==================== 优先级队列操作 (RPriorityQueue) ====================

    /**
     * 获取优先级队列
     *
     * @param queueName 队列名
     * @return 优先级队列对象
     */
    public static <T> RPriorityQueue<T> getPriorityQueue(String queueName) {
        return CLIENT.getPriorityQueue(queueName);
    }

    /**
     * 设置优先级队列比较器
     *
     * @param queueName  队列名
     * @param comparator 比较器
     * @return true=设置成功；false=比较器已存在
     */
    public static <T> boolean setPriorityQueueComparator(String queueName, Comparator<? super T> comparator) {
        RPriorityQueue<T> queue = CLIENT.getPriorityQueue(queueName);
        return queue.trySetComparator(comparator);
    }

    /**
     * 向优先级队列添加元素
     *
     * @param queueName 队列名
     * @param data      数据
     * @return true=添加成功；false=添加失败
     */
    public static <T> boolean addPriorityQueueObject(String queueName, T data) {
        RPriorityQueue<T> queue = CLIENT.getPriorityQueue(queueName);
        return queue.offer(data);
    }

    /**
     * 从优先级队列获取并移除元素
     *
     * @param queueName 队列名
     * @return 优先级最高的元素，如果队列为空则返回null
     */
    public static <T> T pollPriorityQueueObject(String queueName) {
        RPriorityQueue<T> queue = CLIENT.getPriorityQueue(queueName);
        return queue.poll();
    }

    /**
     * 获取优先级队列所有元素
     *
     * @param queueName 队列名
     * @return 所有元素列表
     */
    public static <T> List<T> readAllPriorityQueue(String queueName) {
        RPriorityQueue<T> queue = CLIENT.getPriorityQueue(queueName);
        return queue.readAll();
    }

    /**
     * 从优先级队列尾部移除元素并添加到另一个队列头部
     *
     * @param sourceQueueName 源队列名
     * @param targetQueueName 目标队列名
     * @return 移动的元素，如果源队列为空则返回null
     */
    public static <T> T pollLastAndOfferFirstTo(String sourceQueueName, String targetQueueName) {
        RPriorityQueue<T> queue = CLIENT.getPriorityQueue(sourceQueueName);
        return queue.pollLastAndOfferFirstTo(targetQueueName);
    }

    // ==================== 传输队列操作 (RTransferQueue) ====================

    /**
     * 获取传输队列
     *
     * @param queueName 队列名
     * @return 传输队列对象
     */
    public static <T> RTransferQueue<T> getTransferQueue(String queueName) {
        return CLIENT.getTransferQueue(queueName);
    }

    /**
     * 向传输队列添加元素
     *
     * @param queueName 队列名
     * @param data      数据
     * @return true=添加成功；false=添加失败
     */
    public static <T> boolean addTransferQueueObject(String queueName, T data) {
        RTransferQueue<T> queue = CLIENT.getTransferQueue(queueName);
        return queue.offer(data);
    }

    /**
     * 传输元素到等待的消费者
     *
     * @param queueName 队列名
     * @param data      数据
     * @return true=传输成功；false=没有等待的消费者
     */
    public static <T> boolean transferToConsumer(String queueName, T data) {
        RTransferQueue<T> queue = CLIENT.getTransferQueue(queueName);
        return queue.tryTransfer(data);
    }

    /**
     * 传输元素到等待的消费者（带超时）
     *
     * @param queueName 队列名
     * @param data      数据
     * @param timeout   超时时间
     * @param unit      时间单位
     * @return true=传输成功；false=超时或没有等待的消费者
     */
    public static <T> boolean transferToConsumer(String queueName, T data, long timeout, TimeUnit unit) {
        RTransferQueue<T> queue = CLIENT.getTransferQueue(queueName);
        try {
            return queue.tryTransfer(data, timeout, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }

    /**
     * 获取等待消费者的数量
     *
     * @param queueName 队列名
     * @return 等待消费者数量
     */
    public static int getWaitingConsumerCount(String queueName) {
        RTransferQueue<?> queue = CLIENT.getTransferQueue(queueName);
        return queue.getWaitingConsumerCount();
    }

    /**
     * 检查是否有等待的消费者
     *
     * @param queueName 队列名
     * @return true=有等待的消费者；false=没有等待的消费者
     */
    public static boolean hasWaitingConsumer(String queueName) {
        RTransferQueue<?> queue = CLIENT.getTransferQueue(queueName);
        return queue.hasWaitingConsumer();
    }

    /**
     * 获取传输队列所有元素
     *
     * @param queueName 队列名
     * @return 所有元素列表
     */
    public static <T> List<T> readAllTransferQueue(String queueName) {
        RTransferQueue<T> queue = CLIENT.getTransferQueue(queueName);
        return queue.readAll();
    }

    // ==================== 环形缓冲区操作 (RRingBuffer) ====================

    /**
     * 获取环形缓冲区
     *
     * @param bufferName 缓冲区名
     * @return 环形缓冲区对象
     */
    public static <T> RRingBuffer<T> getRingBuffer(String bufferName) {
        return CLIENT.getRingBuffer(bufferName);
    }

    /**
     * 设置环形缓冲区容量
     *
     * @param bufferName 缓冲区名
     * @param capacity   容量
     * @return true=设置成功；false=容量已设置
     */
    public static boolean setRingBufferCapacity(String bufferName, int capacity) {
        RRingBuffer<?> buffer = CLIENT.getRingBuffer(bufferName);
        return buffer.trySetCapacity(capacity);
    }

    /**
     * 强制设置环形缓冲区容量
     *
     * @param bufferName 缓冲区名
     * @param capacity   容量
     */
    public static void forceSetRingBufferCapacity(String bufferName, int capacity) {
        RRingBuffer<?> buffer = CLIENT.getRingBuffer(bufferName);
        buffer.setCapacity(capacity);
    }

    /**
     * 向环形缓冲区添加元素
     *
     * @param bufferName 缓冲区名
     * @param data       数据
     * @return true=添加成功；false=添加失败
     */
    public static <T> boolean addRingBufferObject(String bufferName, T data) {
        RRingBuffer<T> buffer = CLIENT.getRingBuffer(bufferName);
        return buffer.offer(data);
    }

    /**
     * 从环形缓冲区获取并移除元素
     *
     * @param bufferName 缓冲区名
     * @return 头部元素，如果缓冲区为空则返回null
     */
    public static <T> T pollRingBufferObject(String bufferName) {
        RRingBuffer<T> buffer = CLIENT.getRingBuffer(bufferName);
        return buffer.poll();
    }

    /**
     * 获取环形缓冲区容量
     *
     * @param bufferName 缓冲区名
     * @return 容量
     */
    public static int getRingBufferCapacity(String bufferName) {
        RRingBuffer<?> buffer = CLIENT.getRingBuffer(bufferName);
        return buffer.capacity();
    }

    /**
     * 获取环形缓冲区剩余容量
     *
     * @param bufferName 缓冲区名
     * @return 剩余容量
     */
    public static int getRingBufferRemainingCapacity(String bufferName) {
        RRingBuffer<?> buffer = CLIENT.getRingBuffer(bufferName);
        return buffer.remainingCapacity();
    }

    /**
     * 获取环形缓冲区大小
     *
     * @param bufferName 缓冲区名
     * @return 当前大小
     */
    public static int getRingBufferSize(String bufferName) {
        RRingBuffer<?> buffer = CLIENT.getRingBuffer(bufferName);
        return buffer.size();
    }

    /**
     * 检查环形缓冲区是否为空
     *
     * @param bufferName 缓冲区名
     * @return true=缓冲区为空；false=缓冲区不为空
     */
    public static boolean isRingBufferEmpty(String bufferName) {
        RRingBuffer<?> buffer = CLIENT.getRingBuffer(bufferName);
        return buffer.isEmpty();
    }

    /**
     * 清空环形缓冲区
     *
     * @param bufferName 缓冲区名
     */
    public static void clearRingBuffer(String bufferName) {
        RRingBuffer<?> buffer = CLIENT.getRingBuffer(bufferName);
        buffer.clear();
    }
}
